//operator to prcess the orderby key word in sql
/** Select Operation **/

package qp.operators;

import qp.utils.*;

import java.io.*;
import java.util.*;
import java.lang.*;

public class Orderby extends Operator{

	Operator base; //the base of Orderby should be project operator
	Vector attrSet; //the attribute list followin orderby keyward in SQL

	// number of tuples per outbatch
	int batchsize; 
	
	//input buffer and output buffer required during exexution 
	Batch inbatch;
	Batch outbatch;

	int start;       // Cursor position in the input buffer
	boolean eos;  // Indiacate whether end of stream is reached or not
	
	//get the unique file name to store batch in open()
	static int filenum = 0; 
	
	// The file name where the sorted batch object is stored
//	String rfname;   
	// The ObjectInputStream object, just call this object to readObject in next()
	ObjectInputStream sortedInput;
	
	public Orderby(Operator base, Vector as, int type){
		super(type);
		this.base = base;
		this.attrSet=as;
		}

	//Debug::PPrint(oderby)  need the supprot of 
	//oderby.getBase()
	public Operator getBase()
	{
		return base;
	}

	//RandomOptimizer::makeExecPlan()  
	//the support of Orderby::getBase()  Orderby::setBase()
	public void setBase(Operator base)
	{
		this.base = base;
	}

	//create a unique file name
	private String getUniqueName()
		{
			filenum++; 
			String tempName;
			tempName = "Orderby_" +  String.valueOf(filenum) + ".data";

			return tempName;
		}
	
	public boolean open(){
		eos=false;     // Since the stream is just opened
		start = 0;   // set the cursor to starting position in input buffer
		
		//get the tuples number per outbatch
		int tuplesize = schema.getTupleSize();
		batchsize=Batch.getPageSize()/tuplesize;    /////check when the operator::setscheme will be called for selection, do necessary adjust for oderby 

		//open the base.open()
		if(!base.open())
	    		return false;
		
		//create a unique file name
		String rawBatchFile = this.getUniqueName();
		
		try
		{
			//open a file to stroe the batch
			ObjectOutputStream out = new ObjectOutputStream(new FileOutputStream(rawBatchFile));
			
			//get all the batch from base operator, 
			Batch temp_inbatch;
			while((temp_inbatch = (Batch) base.next()) != null)
			{
				out.writeObject(temp_inbatch);
			}

			//close output file
			out.close();
		}catch (IOException io)
		{
				System.out.println("Oderby:writing the temporary file error");
				return false;
		}


		//external sort for the file accroding to the attribute list
		ExternalSort tempSort = new ExternalSort(rawBatchFile, attrSet, this.schema);	

		String sortedFile = null;
		try{
			//return the String disply the sorted file,  input file ---> sorted file		
			sortedFile = tempSort.sort();   //, 10, 1);
			//sortedFile = tempSort.sort_debug();
		}catch(IOException io)
		{
			System.out.println("externalsort@orderby throw IOException");
		}catch(ClassNotFoundException classio)
		{
			System.out.println("externalsort@orderby throw ClassNotFoundException");
		}

		//means no tuples meed the condition of query
		if(sortedFile == null)
		{
			//let the next return null, by set this flag as true
			eos = true;
		}
		else //open the input stream
		{
			//open the input stream, it will used to get the batch object in next()
			try{
				sortedInput = new ObjectInputStream(new FileInputStream(sortedFile));
			}catch(IOException io)
			{
			    System.err.println("Objectby::open error in reading the file, sortedFile");
			    System.exit(1);
			}
		}
		
		//close the base.close(), because we have read all the data from base oeprator
		if(!base.close())
	    		return false;	
		else
			return true;
		
		
	
		}

/*
	
	//if there is no more tuple can be found in given file, justreturn null
	//else reture nthe tuple associated with highest priority in given file 
	//		delete the tuple in the file
	//note : file indicated by BatchFIle stored the batch object, batch file store the tuples
	private Tuple GetFirstTuple(String BatchFile, Vector attributeList)
	{
		//in put stream
		try{
			ObjectInputStream in;
			in = new ObjectInputStream(new FileInputStream(BatchFile));
		}catch(IOException io){
			System.err.println("Oderby::GetFirstTuple error in reading the file");
			System.exit(1);
			}
		
		//if no tuple can be found return null
		
		//scan the all batch object in Batch file to find the max tuple 
		//mean while keep the batch, associated with max tuple in menmory

		//delete the max tuple in the batch object and write the batch object back to the file

		//returen the max tuple
	}
	
	//return the bool which tuple is associated with higher priority according to the orderby attribute list
	private boolean CompareTuple(Tuple a, Tuple b, Vector OrderbyList)
	{
		return true;
	}


	public Next(){
		outbatch = new Batch(batchsize);

		//eos is true, means there is no more tuples can be get from inbatch
		if(eos)
		{
			close();
			return null;
		}

		int i =0; //the index used in loop
		
		//loop until outbatch is full
		while(!outbatch.isFull())
		{
			//local file ---> max tuple, according to attribute in orderby list
			Tuple temp_tuple = this.GetFirstTuple(rfname, attrSet);

			//check whether meed the end of input stream
			if (temp_tuple == null) 
			{
				eos=true;  //flag the event, meet the end of input stream
				return outbatch;
			}
			else 
			{
				//max tuple --> outbatch
				outbatch.add(temp_tuple); 
			}

		}

		//return outbatch
		return outbatch;
		}
*/		
	public Batch next(){
		outbatch = new Batch(batchsize);

		//eos is true, means there is no more tuples can be get from inbatch
		if(eos)
		{
			close();
			return null;
		}

		int i =0; //the index used in loop
		
		//loop until outbatch is full
		while(!outbatch.isFull())
		{
			//mean need to read a new batch from local file
			if(start==0)
			{
				//read inbatch from  local file produce in open
				//inbatch = (Batch) base.next();  //debug
				try
				{
//					inbatch.clear();//clear the original input buffer
					inbatch = (Batch)sortedInput.readObject(); //read one page into input buffer from file
				} //mean meet the end of input stream
				catch (EOFException e) 
				{
					System.out.println("Orderby meet the end of file");
					eos = true; 	
					return outbatch;
				}
				catch(IOException io){
		    			System.out.println("Orderby:IOException error");
		    			System.exit(1);
				}
				catch(ClassNotFoundException c)
				{
					    System.out.println("Orderby:Some error in deserialization ");
					    System.exit(1);
				}
				
			}

			//loop until outbatch is full/ inbatch is finished
			for(i=start; (i<inbatch.size()) && (! outbatch.isFull()); i++)
			{
				//inbatch-> tuples -> outbatch
				//debug
				Tuple temp_tuple = inbatch.elementAt(i);
				outbatch.add(temp_tuple);
			}

			//matain the cursor
			if(i==inbatch.size())
				start=0;
			else
				start=i;
		}

		//return outbatch
		return outbatch;
		}

	public boolean close()
		{
			return true;
		}

	public Object clone(){
		//copy the base object
		Operator newbase = (Operator) base.clone();

		//copy the relatived Object in current object
		Vector newattr = new Vector();
		for (int i=0; i<attrSet.size(); i++)
				newattr.add((Attribute) ((Attribute)attrSet.elementAt(i)).clone()) ;

		//initialise a new project operator
		Orderby neworder = new Orderby(newbase, newattr,optype);

		//copy the sechme from base to operator
		//newbase -> schme -> newproject
		
		////oderby need all the attribule. 
		//the schema related code should be the same 
		// between RandomInitialPlan::createOrderbyOp , oderby::clone(),  RandomOptimizer::modifySchem(root)
		Schema newSchema = newbase.getSchema();
		neworder.setSchema(newSchema);

		return neworder;
		}
	
}